import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.util.List;

/**
 * @author lilulu
 * @date 2023/2/20 15:03
 */
public class hbase_api_test01 {
    /**
     * 1) 通过HBase的连接工厂类构建连接对象
     * 2) 根据连接对象, 获取相关的管理对象:  admin(执行对表进行操作)  table(执行对表数据的操作)
     * 3) 执行相关的操作
     * 4) 处理结果集 (只有查询才有结果集)
     * 5) 释放资源
     */

    //创建表
    @Test
    public void test01() throws Exception{

//        1) 通过HBase的连接工厂类构建连接对象
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
        Connection connection = ConnectionFactory.createConnection(configuration);
        //2) 根据连接对象, 获取相关的管理对象:  admin(执行对表进行操作)  table(执行对表数据的操作)
        Admin admin = connection.getAdmin();

//        3) 执行相关的操作
        boolean flag = admin.tableExists(TableName.valueOf("WATER_BILL"));
        if (!flag) {
            // 说明表不存在, 需要构建表
            //3.2 创建表
            //3.2.1 创建表的构建器对象
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf("WATER_BILL"));
            //3.2.2 在构建器对象中, 设置表的列族信息
            ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder("C1".getBytes()).build();
            tableDescriptorBuilder.setColumnFamily(familyDescriptor);
            // 3.2.3 得到表结构对象
            TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
            admin.createTable(tableDescriptor);
        }
//        4) 处理结果集 (只有查询才有结果集)
//        5) 释放资源
        admin.close();
    }

    //添加数据
    @Test
    public void test02() throws Exception {
//        1) 通过HBase的连接工厂类构建连接对象
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
        Connection connection = ConnectionFactory.createConnection(configuration);
        //2) 根据连接对象, 获取相关的管理对象:  admin(执行对表进行操作)  table(执行对表数据的操作)
//        Admin admin = connection.getAdmin();
        Table table = connection.getTable(TableName.valueOf("WATER_BILL"));
        Put put = new Put("4944191".getBytes());
        put.addColumn("C1".getBytes(), "NAME".getBytes(), "登卫红".getBytes());
        put.addColumn("C1".getBytes(), "ADDRESS".getBytes(), "贵州省铜仁市德江县7单元267室".getBytes());
        put.addColumn("C1".getBytes(), "SEX".getBytes(), "男".getBytes());
        table.put(put);
        table.close();
        connection.close();
    }

    /**
     * 公共代码抽取
     */
    private Connection connection;
    private Table table;
    private Admin admin;
    @BeforeEach
    public void before() throws Exception{
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
        connection = ConnectionFactory.createConnection(configuration);
        //2) 根据连接对象, 获取相关的管理对象:  admin(执行对表进行操作)  table(执行对表数据的操作)
         admin = connection.getAdmin();
         table = connection.getTable(TableName.valueOf("WATER_BILL"));
    }



    //查询某一条数据
    @Test
    public void test03() throws Exception{
        Get get = new Get("4944191".getBytes());
        Result result = table.get(get);
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            // 4.2 遍历每一个单元格: 一个单元格里面主要包含(rowkey信息, 列族信息, 列名信息, 列值信息)
            byte[] rowArray = cell.getRowArray();
            byte[] familyArray = cell.getFamilyArray();
            byte[] cellQualifierArray = cell.getQualifierArray();
            byte[] valueArray = cell.getValueArray();
            String rowKey = Bytes.toString(rowArray);
            String family = Bytes.toString(familyArray);
            String columnName = Bytes.toString(cellQualifierArray);
            String value = Bytes.toString(valueArray);
            System.out.println("rowkey为:"+rowKey +", 列族为:"+family +"; 列名为:"+columnName+"; 列值为:"+value);
        }
    }

    @Test
    public void test04() throws Exception{
        // 3- 执行相关的操作
        Get get = new Get("4944191".getBytes());

        Result result = table.get(get);  // 一个 result对象表示一行数据
        //4- 处理结果集
        // 4.1  将一行中每一个单元格获取
        List<Cell> cells = result.listCells();

        // 4.2 遍历每一个单元格: 一个单元格里面主要包含(rowkey信息, 列族信息, 列名信息, 列值信息)
        for (Cell cell : cells) {
            byte[] rowKeyBytes = CellUtil.cloneRow(cell);
            byte[] familyBytes = CellUtil.cloneFamily(cell);
            byte[] columnNameBtyes = CellUtil.cloneQualifier(cell);
            byte[] valueBytes = CellUtil.cloneValue(cell);

            String rowKey = Bytes.toString(rowKeyBytes);
            String family = Bytes.toString(familyBytes);
            String columnName = Bytes.toString(columnNameBtyes);
            String value = Bytes.toString(valueBytes);

            System.out.println("rowkey为:"+rowKey +", 列族为:"+family +"; 列名为:"+columnName+"; 列值为:"+value);

        }

    }


    //删除数据
    @Test
    public void test05() throws Exception{
        Delete delete = new Delete("4944191".getBytes());
        table.delete(delete);
    }


    //删除表
    @Test
    public void test06() throws Exception{
        //首先需要判断表是否被禁用
        if (admin.isTableEnabled(TableName.valueOf("WATER_BILL"))) {
            admin.disableTable(TableName.valueOf("WATER_BILL"));
        }
        admin.deleteTable(TableName.valueOf("WATER_BILL"));
    }


    @Test
    public void test07() throws Exception{
        /**
         *  需求: 查询2020年 6月份所有用户的用水量:
         *         日期字段: RECORD_DATE
         *         用水量: NUM_USAGE
         *         用户: NAME
         */
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilter1 = new SingleColumnValueFilter(
                "C1".getBytes(),
                "RECORD_DATE".getBytes(),
                CompareOperator.GREATER_OR_EQUAL,
                new BinaryComparator("2020-06-01".getBytes())
        );

        SingleColumnValueFilter singleColumnValueFilter2 = new SingleColumnValueFilter(
                "C1".getBytes(),
                "RECORD_DATE".getBytes(),
                CompareOperator.LESS_OR_EQUAL,
                new BinaryComparator("2020-06-30".getBytes())
        );
        //3.1.2 构建 filter集合, 将filter合并在一起
        FilterList filterList = new FilterList();
        filterList.addFilter(singleColumnValueFilter1);
        filterList.addFilter(singleColumnValueFilter2);
        scan.setFilter(filterList);

        scan.setLimit(10);

        scan.addColumn("C1".getBytes(), "NAME".getBytes());
        scan.addColumn("C1".getBytes(), "NUM_USAGE".getBytes());
        scan.addColumn("C1".getBytes(), "RECORD_DATE".getBytes());

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                byte[] bytes = CellUtil.cloneQualifier(cell);
                String columnName = Bytes.toString(bytes);

                byte[] rowKeyBytes = CellUtil.cloneRow(cell);
                byte[] familyBytes = CellUtil.cloneFamily(cell);
                byte[] valueBytes = CellUtil.cloneValue(cell);

                String rowKey = Bytes.toString(rowKeyBytes);
                String family = Bytes.toString(familyBytes);

                Object value;
                if ("NUM_USAGE".equals(columnName)) {
                    value = Bytes.toDouble(valueBytes);
                }else{
                    value = Bytes.toString(valueBytes);
                }

                System.out.println("rowkey为:"+rowKey +", 列族为:"+family +"; 列名为:"+columnName+"; 列值为:"+value);
            }

            System.out.println("-------------------");
        }
    }


    @AfterEach
    public void after() throws Exception{
        admin.close();
        table.close();
        connection.close();
    }
}
